package com.hoki.zj.config;

import com.hoki.zj.Listener.MyJobListener;
import com.hoki.zj.Listener.MyReaderListener;
import com.hoki.zj.Listener.MyWriterListener;
import com.hoki.zj.batch.domain.BlogInfo;
import com.hoki.zj.batch.tasklet.MyTasklet;
import com.hoki.zj.utils.MyBeanValidator;
import com.hoki.zj.utils.MyItemProcessor;
import com.hoki.zj.utils.MyItemProcessorNew;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.batch.MyBatisCursorItemReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;

/**
 * SpringBatch Config
 * 1.JobRepositoryの定義
 * 2.JobLauncher
 * 3.Job
 * 4.Step:
 *  4.1:ItemReader
 *  4.2:ItemProcessor
 *  4.3:ItemWriter
 *  4.4:Step設定
 */
@Configuration
public class MyBatchConfig {
    private Logger logger= LoggerFactory.getLogger(MyBatchConfig.class);

    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @Autowired
    private final MyTasklet myTasklet;

    public MyBatchConfig(MyTasklet myTasklet) {
        this.myTasklet = myTasklet;
    }

    /**
     * 1.JobRepositoryの定義：Jobの登録コンテナ、DBとの交互（Transactionなど）
     * @param dataSource　dataSource
     * @param transactionManager　transactionManager
     * @return JobRepository
     * @throws Exception　exception
     */
    @Bean
    public JobRepository myJobRepository(@Qualifier("dataSource") DataSource dataSource,
                                         PlatformTransactionManager transactionManager) throws Exception
    {
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDatabaseType("mysql");
        jobRepositoryFactoryBean.setTransactionManager(transactionManager);
        jobRepositoryFactoryBean.setDataSource(dataSource);
        return jobRepositoryFactoryBean.getObject();
    }

    /**
     * 2.JobLauncherの定義：Jobの起動、JobRepositoryと結びつく
     * @param dataSource dataSource
     * @param transactionManager transactionManager
     * @return jobLauncher
     * @throws Exception exception
     */
    @Bean
    public SimpleJobLauncher myJobLauncher(@Qualifier("dataSource") DataSource dataSource,
                                           PlatformTransactionManager transactionManager) throws Exception
    {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        // JobRepositoryと結びつく
        jobLauncher.setJobRepository(myJobRepository(dataSource, transactionManager));
        return jobLauncher;
    }

    /**
     * 3.Jobの定義
     * @param jobBuilderFactorys jobBuilderFactorys
     * @param myStep Job執行のStep
     * @return Job
     */
    @Bean
    public Job myJob(JobBuilderFactory jobBuilderFactorys, Step myStep) {
        return jobBuilderFactorys.get("myJob")
                .incrementer(new RunIdIncrementer()) // 増量
                .flow(myStep) // ステップ
                .end()
                .listener(myJobListener()) // JobListener
                .build();
    }
    /**
     * 3.1 JobListenerの取得
     * @return MyJobListener
     */
    @Bean
    public MyJobListener myJobListener() {
        return new MyJobListener();
    }

    /**
     * 4.Step定義
     * 4.1 ItemReaderの定義：dataの読み込み、domainとの関連付け
     */
    @Bean
    public ItemReader<BlogInfo> reader() {
        // FlatFileItemReaderを用いて、cvsファイルを読み込む、一行ごとに
        FlatFileItemReader<BlogInfo> reader = new FlatFileItemReader<>();
        // ファイルのルートを設定する
        reader.setResource(new ClassPathResource("static/bloginfo.csv"));
        // domainとの関連付け
        reader.setLineMapper(new DefaultLineMapper<BlogInfo>(){
            {
                setLineTokenizer(new DelimitedLineTokenizer() {
                    {   // cvsファイルの中に、対応するフィールド
                        setNames(new String[]{"blogAuthor", "blogUrl", "blogTitle", "blogItem"});
                    }
                });
                setFieldSetMapper(new BeanWrapperFieldSetMapper<BlogInfo>() {
                    {
                        setTargetType(BlogInfo.class);
                    }
                });
            }
        });
        return reader;
    }

    /**
     * 4.2 ItemProcessorの定義:data処理、dataの校正
     * @return ItemProcessor
     */
    @Bean
    public ItemProcessor<BlogInfo, BlogInfo> processor() {
        MyItemProcessor myItemProcessor = new MyItemProcessor();
        // バリデーターを設定する
        myItemProcessor.setValidator(myBeanValidator());
        return myItemProcessor;
    }
    /**
     * 4.2.1 バリデーターの定義
     * JSR-303のValidator
     * @return MyBeanValidator
     */
    @Bean
    public MyBeanValidator  myBeanValidator() {
        return new MyBeanValidator<BlogInfo>();
    }

    /**
     * 4.3 ItemWriterの定義
     * @param dataSource　dataSource
     * @return
     */
    @Bean
    public ItemWriter<BlogInfo> writer(@Qualifier("dataSource") DataSource dataSource) {
        //　JdbcBatchItemWriterを使って、DBにdataを入力する
        JdbcBatchItemWriter<BlogInfo> writer = new JdbcBatchItemWriter<>();
        // パラメータがあるSqlを設定する
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        String sql = "insert into bloginfo "+" (blogAuthor,blogUrl,blogTitle,blogItem) "
                +" values(:blogAuthor,:blogUrl,:blogTitle,:blogItem)";
        writer.setSql(sql);
        writer.setDataSource(dataSource);
        return writer;
    }

    /**
     * 4.4 ItemReader, ItemProcessor, ItemWriterをStepと結びつく
     * @param stepBuilderFactory stepBuilderFactory
     * @param reader reader
     * @param writer writer
     * @param processor processor
     * @return Step
     */
    @Bean
    public Step myStep(StepBuilderFactory stepBuilderFactory, ItemReader<BlogInfo> reader,
                       ItemWriter<BlogInfo> writer, ItemProcessor<BlogInfo, BlogInfo> processor){
        return stepBuilderFactory
                .get("myStep")
                // Chunkモデル(dataを一つずつ読み込んで、処理する。一定の量に達したら、一気にwriterを使ってDBに入力する)
                .<BlogInfo, BlogInfo>chunk(65000)
                // readerを設定する、retry回数を設定する、見逃せる失敗の回数を設定する
                .reader(reader).faultTolerant().retryLimit(3).retry(Exception.class).skip(Exception.class).skipLimit(2)
                .listener(new MyReaderListener())
                .processor(processor)
                .writer(writer).faultTolerant().skip(Exception.class).skipLimit(2)
                .listener(new MyWriterListener())
                .build();
    }

    //==================================DBからdataを読み込む==================================//
    @Bean
    public Job myJobNew(JobBuilderFactory jobBuilderFactory, Step myStepNew) {
        return jobBuilderFactory.get("myJobNew")
                .incrementer(new RunIdIncrementer()) // 増量
                .flow(myStepNew) // ステップ
                .end()
                .listener(myJobListener()) // JobListener
                .build();
    }

    @Bean
    public Step myStepNew(StepBuilderFactory stepBuilderFactory, MyBatisCursorItemReader<BlogInfo> readerNew,
                          ItemWriter<BlogInfo> writerNew, ItemProcessor<BlogInfo, BlogInfo> processorNew) {
        return stepBuilderFactory
                .get("myStepNew")
                // Chunkルール(dataを一つずつ読み込んで、処理する。一定の量に達したら、一気にwriterを使ってDBに入力する)
                .<BlogInfo, BlogInfo>chunk(65000)
                // readerを設定する、retry回数を設定する、見逃せる失敗の回数を設定する
                .reader(readerNew).faultTolerant().retryLimit(3).retry(Exception.class).skip(Exception.class).skipLimit(2)
                .listener(new MyReaderListener())
                .processor(processorNew)
                .writer(writerNew).faultTolerant().skip(Exception.class).skipLimit(2)
                .listener(new MyWriterListener())
                .build();
    }

    /**
     * 新たなItemProcessorを定義する:data処理、dataの校正
     * @return csvItemProcessor
     */
    @Bean
    public ItemProcessor<BlogInfo, BlogInfo> processorNew(){
        MyItemProcessorNew csvItemProcessor = new MyItemProcessorNew();
        // バリデーターを設定する
        csvItemProcessor.setValidator(myBeanValidator());
        return csvItemProcessor;
    }

    /**
     * ItemReaderの定義：DBからdataを読み込む
     * @param authorId　authorId
     * @return reader
     */
    @Bean
    @StepScope
    //Spring Batchが提供されたクラスBeanScope
    // （StepScope:自分で定義するSpring bean scopeとして機能する）。
    // 役割は各Batchesと繋がること。
    // Springの中に配置されたBeanをsteps実行開始の時にインスタンスすることを認めて、configとパラメータも設定する
    // @Value JobのjobParametersから能動的にパラメータを渡し、sqlを能動的に実行する
    public MyBatisCursorItemReader<BlogInfo> itemReaderNew(@Value("#{jobParameters[authorId]}") String authorId) {
        System.out.println("DB読み込みスタート");
        MyBatisCursorItemReader<BlogInfo> reader = new MyBatisCursorItemReader<>();
        // DBからdataを読み込み中にsqlが実行されたmapperでのメッソドを設定する
        reader.setQueryId("com.hoki.zj.batch.mapper.BlogInfoMapper.queryInfoById");

        reader.setSqlSessionFactory(sqlSessionFactory);
        Map<String , Object> map = new HashMap<>();
        // Sqlに渡すパラメータを設定する: #{XXX}
        map.put("authorId" , Integer.valueOf(authorId));
        reader.setParameterValues(map);
        return reader;
    }


    /**
     * ItemWriter定义：指定datasource，设置批量插入sql语句，写入数据库
     * @param dataSource dataSource
     * @return writer
     */
    @Bean
    public ItemWriter<BlogInfo> writerNew(@Qualifier("dataSource")  DataSource dataSource){
        // DBにデータを追加する
        JdbcBatchItemWriter<BlogInfo> writer = new JdbcBatchItemWriter<>();
        // パラメータがあるsqlを設定する
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        String sql = "insert into bloginfonew2 "+" (blogAuthor,blogUrl,blogTitle,blogItem) "
                +" values(:blogAuthor,:blogUrl,:blogTitle,:blogItem)";
        writer.setSql(sql);
        writer.setDataSource(dataSource);
        return writer;
    }

    //==================== ====================
    @Bean
    public Job jobTasklet(JobBuilderFactory jobBuilderFactory, Step myStep3) {
        return jobBuilderFactory.get("jobTa")
                .incrementer(new RunIdIncrementer()) // 増量
                .flow(myStep3) // ステップ
                .end()
                .listener(myJobListener()) // JobListener
                .build();
    }

    @Bean
    public Step myStep3(StepBuilderFactory stepBuilderFactory) {
        return stepBuilderFactory.get("myStep3")
                .tasklet(myTasklet)
                .build();
    }
}
